"""
Case Type   : 故障&可靠性测试
Case Name   : 事务提交方式为remote_apply，非最大可用模式下，主机执行事务过程中备节点注入磁盘满故障
Create At   : 2021/01/27
Owner       : @peilinqian
Description :
    step1:创建表;
    step2:设置guc参数;
    step3:查询集群同步方式;
    step4:等待主备一致;
    step5:插入数据;
    step6:步骤4执行过程中，sync备节点注入磁盘满;
    step7:查询潜在同步备同步情况;
    step8:清除故障，恢复故障节点;
    step9:查询所有备节点同步情况;
    step10:主机插入数据;
    step11:步骤10执行过程中，sync备节点注入磁盘满;
    step12:等待15s清除故障，并恢复环境;
    step13:查询备节点同步情况;
    step14:设置同步备为ANY 2，主机插入数据;
    step15:步骤13过程中，备节点均注入故障;
    step16:清除故障，并恢复环境;
    step17:查询备节点同步情况;
Expect      :
    step1:创建表; expect：成功
    step2:设置guc参数; expect：成功
    step3:查询集群同步方式; expect：正确
    step4:等待主备一致; expect：正确
    step5:插入数据; expect：1min后insert成功
    step6:步骤4执行过程中，sync备节点注入磁盘满; expect：成功
    step7:查询潜在同步备同步情况; expect：同步成功
    step8:清除故障，恢复故障节点; expect：清除成功
    step9:查询所有备节点同步情况; expect：同步成功
    step10:主机插入数据; expect：备节点恢复后插入成功
    step11:步骤10执行过程中，sync备节点注入磁盘满; expect：成功
    step12:等待15s清除故障，并恢复环境; expect：成功
    step13:查询备节点同步情况; expect：同步成功
    step14:设置同步备为ANY 2; expect：卡死无返回
    step15:步骤13过程中，备节点均注入故障; expect：成功
    step16:清除故障，并恢复环境; expect：成功
    step17:查询备节点同步情况; expect：同步成功
History     :
    Modified by peilinqian at 2024-8-24:executDbSql公共方法优化后，入参调整
    Modified by peilinqian at 2024-9-2:优化用例，非最大可用模式下开启场景下执行该用例
"""

import datetime
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonSH import RestartDbCluster
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class RecoveryDelay(unittest.TestCase):
    pri_sh = CommonSH('PrimaryDbUser')
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')

    @RestartDbCluster(*nodes_tuple)
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_root = Node(node='PrimaryRoot')
        self.pri_dbuser = Node(node='PrimaryDbUser')
        self.comsta = [Common('Standby1Root'), Common('Standby2Root')]
        self.sta_root = [Node(node='Standby1Root'),
                         Node(node='Standby2Root')]
        self.sta_sh = [CommonSH('Standby1DbUser'),
                       CommonSH('Standby2DbUser')]
        self.constant = Constant()
        self.db_disk_name = []
        self.tb_name = 't_dbsys241'
        self.conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      macro.DB_PG_CONFIG_NAME)
        self.pri_hostname = self.pri_root.sh('hostname').result()

        self.log.info('----同步集群时间----')
        current = self.pri_root.sh(
            "date \"+%m/%d/%Y %H:%M:%S\"").result()
        self.log.info(current)
        datecmd = f'date -s "{current}"'
        self.log.info(datecmd)
        for i in range(2):
            result = self.sta_root[i].sh(datecmd).result()
            self.log.info(result)

        self.log.info('================获取备节点磁盘名=========')
        for i in range(2):
            self.db_disk_name.append(
                self.comsta[i].get_disk_name(macro.DB_INSTANCE_PATH))

    def test_recoverydelay(self):
        step_txt = '----step0:查看synchronous_commit初始配置值;----'
        self.log.info(step_txt)
        self.init_para1 = self.pri_sh.show_param("synchronous_commit")
        self.init_para2 = self.pri_sh.show_param("synchronous_standby_names")
        self.init_para3 = self.pri_sh.show_param("recovery_min_apply_delay")
        self.init_para4 = self.pri_sh.show_param("most_available_sync")

        step_txt = '----step1:创建表; expect：成功----'
        self.log.info(step_txt)
        sql = f"drop table if exists {self.tb_name};" \
            f"create table {self.tb_name}(i int, s char(10));"
        result = self.pri_sh.executDbSql(sql)
        self.log.info(result)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result)

        step_txt = '----step2:设置guc参数; expect：成功----'
        self.log.info(step_txt)
        self.log.info('----设置synchronous_commit=remote_apply----')
        result = self.pri_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            'synchronous_commit=remote_apply')
        self.assertTrue(result, '执行失败:' + step_txt)
        self.log.info('---设置synchronous_standby_names='
                      'FIRST 1(dn_6002,dn_6003)--')
        result = self.pri_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            "synchronous_standby_names='FIRST 1(dn_6002,dn_6003)'",
            self.pri_hostname)
        self.assertTrue(result, '执行失败:' + step_txt)
        self.log.info('----设置recovery_min_apply_delay=1min----')
        result = self.pri_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            'recovery_min_apply_delay=1min')
        self.assertTrue(result, '执行失败:' + step_txt)
        self.log.info('----设置most_available_sync=off----')
        result = self.pri_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            'most_available_sync=off')
        self.assertTrue(result, '执行失败:' + step_txt)

        self.log.info('----查询参数----')
        result = self.pri_sh.show_param("synchronous_commit")
        self.log.info(result)
        self.assertIn('remote_apply', result, '执行失败:' + step_txt)
        result = self.pri_sh.show_param('recovery_min_apply_delay')
        self.log.info(result)
        self.assertIn('1min', result, '执行失败:' + step_txt)
        result = self.pri_sh.show_param('synchronous_standby_names')
        self.log.info(result)
        self.assertIn(f"{macro.DN_NODE_NAME.split('/')[1]},"
                      f"{macro.DN_NODE_NAME.split('/')[2]}", result,
                      '执行失败:' + step_txt)
        result = self.pri_sh.show_param('most_available_sync')
        self.log.info(result)
        self.assertIn("off", result, '执行失败:' + step_txt)

        step_txt = '----step3:查询集群同步方式; expect：正确----'
        self.log.info(step_txt)
        sql = "select * from pg_stat_replication;"
        result = self.pri_sh.executDbSql(sql)
        self.log.info(result)
        self.assertIn('Sync', result, '执行失败:' + step_txt)
        self.assertIn('Potential', result, '执行失败:' + step_txt)

        step_txt = '----step4:等待主备一致; expect：正确----'
        self.log.info(step_txt)
        for i in range(2):
            result = self.sta_sh[i].check_data_consistency()
            self.assertTrue(result, '执行失败:' + step_txt)

        step5_txt = '----step5:插入数据; expect：1min后insert成功----'
        self.log.info(step5_txt)
        sql = f"select sysdate;insert into {self.tb_name} values(1,'test');" \
            f"select sysdate;"
        insert_thread = ComThread(self.pri_sh.executDbSql,
                                  args=(sql,))
        insert_thread.setDaemon(True)
        insert_thread.start()
        start = datetime.datetime.now()
        self.log.info("开始插入时间：" + str(start))

        step_txt = '----step6:步骤4执行过程中，sync备节点注入磁盘满; expect：成功----'
        self.log.info(step_txt)
        cfe_cmd = f'{macro.CFE_PATH}/cfe ' \
            f'"inject rfile_full (diskname) values ({self.db_disk_name[0]})"'
        self.log.info(cfe_cmd)
        cfe_result = self.sta_root[0].sh(cfe_cmd).result()
        self.log.info(cfe_result)
        self.assertIn(self.constant.CFE_DISK_FULL_SUCCESS_MSG, cfe_result,
                      '执行失败:' + step_txt)

        self.log.info('========获取insert结果======')
        insert_thread.join(80)
        result = insert_thread.get_result()
        self.log.info(result)
        end = datetime.datetime.now()
        self.log.info("结束插入时间：" + str(end))
        execute_time = (end - start).seconds
        self.log.info(f"execute_time is {execute_time}")
        self.assertIn('INSERT', result)
        self.assertLessEqual(60, execute_time, '执行失败:' + step5_txt)

        step_txt = '----step7:查询潜在同步备同步情况; expect：同步成功----'
        self.log.info(step_txt)
        time.sleep(2)
        sql = f"select sysdate;select * from {self.tb_name};"
        result = self.sta_sh[1].executDbSql(sql)
        self.log.info(result)
        self.assertIn('1 | test', result, '执行失败:' + step_txt)

        step_txt = '----step8:清除故障，恢复故障节点; expect：清除成功----'
        self.log.info(step_txt)
        cfe_cmd = f'{macro.CFE_PATH}/cfe "clean rfile_full ' \
            f'where(diskname={self.db_disk_name[0]})"'
        cfe_result = self.sta_root[0].sh(cfe_cmd).result()
        self.log.info(cfe_result)
        self.assertIn(self.constant.CFE_DISK_CLEAN_SUCCESS_MSG, cfe_result,
                      '执行失败:' + step_txt)
        self.log.info('----启动数据库----')
        self.pri_sh.getDbClusterStatus('status')
        result = self.pri_sh.startDbCluster()
        self.assertTrue(result, '执行失败:' + step_txt)

        step_txt = '----step9:查询所有备节点同步情况; expect：同步成功----'
        self.log.info(step_txt)
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql)
            self.log.info(result)
            self.assertIn('1 | test', result, '执行失败:' + step_txt)

        step10_txt = '----step10:主机插入数据; expect：备节点恢复后插入成功----'
        self.log.info(step10_txt)
        self.log.info('----插入数据----')
        sql = f"select sysdate;insert into {self.tb_name} values(2,'test');" \
            f"select sysdate;"
        insert_thread = ComThread(self.pri_sh.executDbSql, args=(sql,))
        insert_thread.setDaemon(True)
        insert_thread.start()
        start = datetime.datetime.now()
        self.log.info("开始插入时间：" + str(start))

        step_txt = '----step11:步骤10执行过程中，sync备节点注入磁盘满; expect：成功----'
        self.log.info(step_txt)
        cfe_cmd = f'{macro.CFE_PATH}/cfe ' \
            f'"inject rfile_full (diskname) values  ({self.db_disk_name[0]})"'
        self.log.info(cfe_cmd)
        cfe_result = self.sta_root[0].sh(cfe_cmd).result()
        self.log.info(cfe_result)
        self.assertIn(self.constant.CFE_DISK_FULL_SUCCESS_MSG, cfe_result,
                      '执行失败:' + step_txt)

        step_txt = '----step12:等待15s清除故障，并恢复环境; expect：成功----'
        self.log.info(step_txt)
        time.sleep(15)
        cfe_cmd = f'''{macro.CFE_PATH}/cfe "clean  
                        rfile_full  where(diskname={self.db_disk_name[0]})"'''
        cfe_result = self.sta_root[0].sh(cfe_cmd).result()
        self.log.info(cfe_result)
        self.assertIn(self.constant.CFE_DISK_CLEAN_SUCCESS_MSG, cfe_result,
                      '执行失败:' + step_txt)
        self.log.info('----启动数据库----')
        self.pri_sh.getDbClusterStatus('status')
        result = self.pri_sh.startDbCluster()
        self.assertTrue(result, '执行失败:' + step_txt)

        self.log.info('========获取insert结果======')
        insert_thread.join(80)
        result = insert_thread.get_result()
        self.log.info(result)
        end = datetime.datetime.now()
        self.log.info("结束插入时间：" + str(end))
        execute_time = (end - start).seconds
        self.log.info(f"execute_time is {execute_time}")
        self.assertIn(self.constant.insert_success_msg, result,
                      '执行失败:' + step10_txt)

        step_txt = '----step13:查询备节点同步情况; expect：同步成功----'
        self.log.info(step_txt)
        self.log.info('----查询备节点----')
        select_sql = f"select sysdate;select * from {self.tb_name};"
        if execute_time < 60:
            result = self.sta_sh[0].executDbSql(select_sql)
            self.log.info(result)
            self.assertIn('2 | test', result, '执行失败:' + step_txt)
            result = self.sta_sh[1].executDbSql(select_sql)
            self.log.info(result)
            self.assertNotIn('2 | test', result, '执行失败:' + step_txt)
        time.sleep(65 - execute_time)
        for i in range(2):
            result = self.sta_sh[i].executDbSql(select_sql)
            self.log.info(result)
            self.assertIn('2 | test', result, '执行失败:' + step_txt)

        step14_txt = '----step14:设置同步备为ANY 2，主机插入数据; expect：卡死无返回----'
        self.log.info(step14_txt)
        result = self.pri_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            "synchronous_standby_names='ANY 2(dn_6002,dn_6003)'",
            self.pri_hostname)
        self.assertTrue(result, '执行失败:' + step_txt)
        sql = f"select sysdate;" \
            f"insert into {self.tb_name} values (generate_series(3,1000)," \
            f"'test'); select sysdate;"
        update_thread = ComThread(self.pri_sh.executDbSql, args=(sql,))
        update_thread.setDaemon(True)
        update_thread.start()

        step_txt = '----step15:步骤13过程中，备节点均注入故障; expect：成功----'
        self.log.info(step_txt)
        for i in range(2):
            cfe_cmd = f'{macro.CFE_PATH}/cfe ' \
                f'"inject  rfile_full (diskname) ' \
                f'values  ({self.db_disk_name[i]})"'
            self.log.info(cfe_cmd)
            cfe_result = self.sta_root[i].sh(cfe_cmd).result()
            self.log.info(cfe_result)
            self.assertIn(self.constant.CFE_DISK_FULL_SUCCESS_MSG, cfe_result,
                          '执行失败:' + step_txt)

        self.log.info('========获取update结果======')
        update_thread.join(80)
        result = update_thread.get_result()
        self.log.info(result)
        self.pri_sh.getDbClusterStatus('status')
        self.assertEqual(None, result, '执行失败:' + step14_txt)

        step_txt = '----step16:清除故障，并恢复环境; expect：成功----'
        self.log.info(step_txt)
        for i in range(2):
            cfe_cmd = f'{macro.CFE_PATH}/cfe "clean rfile_full ' \
                f'where(diskname={self.db_disk_name[i]})"'
            cfe_result = self.sta_root[i].sh(cfe_cmd).result()
            self.log.info(cfe_result)
            self.assertIn(self.constant.CFE_DISK_CLEAN_SUCCESS_MSG,
                          cfe_result, '执行失败:' + step_txt)
        self.log.info('----重启数据库----')
        self.pri_sh.getDbClusterStatus('status')
        result = self.pri_sh.startDbCluster()
        self.assertTrue(result, '执行失败:' + step_txt)

        step_txt = '----step17:查询备节点同步情况; expect：同步成功----'
        self.log.info(step_txt)
        sql = f"select count(*) from {self.tb_name};"
        for i in range(2):
            result = self.sta_sh[i].executDbSql(sql)
            self.log.info(result)
            self.assertIn('1000', result, '执行失败:' + step_txt)

    def tearDown(self):
        self.log.info('-----环境清理----')
        self.log.info('----清除故障----')
        for i in range(2):
            cfe_cmd = f'''{macro.CFE_PATH}/cfe "clean 
                    rfile_full  where(diskname={self.db_disk_name[i]})"'''
            cfe_result = self.sta_root[i].sh(cfe_cmd).result()
            self.log.info(cfe_result)
        if not self.pri_sh.getDbClusterStatus('status'):
            self.log.info('----清除故障后，重启数据库----')
            self.pri_sh.startDbCluster()

        step1_txt = '----还原参数; expect：还原成功----'
        self.log.info(step1_txt)
        reset_msg1 = self.pri_sh.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'synchronous_commit={self.init_para1}')
        reset_msg2 = self.pri_sh.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f"synchronous_standby_names='{self.init_para2}'",
            self.pri_hostname)
        reset_msg3 = self.pri_sh.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'recovery_min_apply_delay={self.init_para3}')
        reset_msg4 = self.pri_sh.executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'most_available_sync={self.init_para4}')

        step2_txt = '----还原参数后重启数据库; expect：还原成功----'
        self.log.info(step2_txt)
        stop_result = self.pri_sh.stopDbCluster()
        self.log.info(stop_result)
        start_result = self.pri_sh.startDbCluster()
        self.log.info(start_result)

        step3_txt = '----删除表; expect：删除成功----'
        self.log.info(step3_txt)
        sql = f"drop table if exists {self.tb_name};"
        drop_result = self.pri_sh.executDbSql(sql)
        self.log.info(drop_result)

        self.assertTrue(reset_msg1, '执行失败:' + step1_txt)
        self.assertTrue(reset_msg2, '执行失败:' + step1_txt)
        self.assertTrue(reset_msg3, '执行失败:' + step1_txt)
        self.assertTrue(reset_msg4, '执行失败:' + step1_txt)
        self.assertTrue(stop_result, '执行失败:' + step2_txt)
        self.assertTrue(start_result, '执行失败:' + step2_txt)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_result,
                      '执行失败:' + step3_txt)

        self.log.info(f"-----{os.path.basename(__file__)} end-----")
